
#include "manager.h"

void *rp_rpointd_manager(void *arg)
{
	char fname[] = "rp_rpointd_worker";
	char msg[] = "fatal error";
	Err err;
	FLOG_IN(fname);
	
	struct event_base *evb_manager;
	struct event ev_io;
	struct sockaddr_in addr_io_server, addr_hsk_client;
	int fd_hsk_server, fd_hsk_client, fd_io_server, fd_io_client;
	uint16_t random_port;
	int length;

	/*	setup tcp servers    */
	err = rp_comm_tcp_server_init(&fd_hsk_server, &config.addr_hsk, COMM_BLOCK);	// TCP, blocking
	err = rp_comm_tcp_server_init(&fd_io_server,  &config.addr_io,  COMM_NONBLOCK);	// use random port number, TCP, Non-blocking
	if ((random_port = ntohs(config.addr_io.sin_port)) == 0)
	{
		getsockname(fd_io_server, (struct sockaddr *)&addr_io_server, &length);
		random_port = ntohs(addr_io_server.sin_port);
	}
	
	evb_manager = event_init();
	while (rp_score_get_code(global.score) != OP_SHUTDOWN)
	{		
		/*	waiting for the client & send back the I/O port back to the client    */
		err = rp_mutex_lock(&(global.mutex_hsk));
		fd_hsk_client = rp_comm_accept(fd_hsk_server, &addr_hsk_client, sizeof(addr_hsk_client));
		err = rp_mutex_unlock(&(global.mutex_hsk));
		err = rp_comm_send(fd_hsk_client, (void*)&random_port, sizeof(uint16_t));	
		err = rp_score_set_state(global.score, SS_BUSY);
		rp_log(LOG_DEBUG, fname, "Handshake with the client from %s:%d");		// ip:port

		event_set(&ev_io, fd_io_server, EV_READ|EV_PERSIST, rp_rpointd_manager_io, (void *)&fd_hsk_client);
		event_base_set(evb_manager, &ev_io);		
		if (event_add(&ev_io, &(config.tv_io_timeout)) < 0)
		{	
			rp_log(LOG_FATAL, fname, msg);
			return NULL;
		}		
		event_base_dispatch(evb_manager);
		
		close(fd_hsk_client);
	}

	/*	Shutdown the tcp servers    */
	err = rp_comm_tcp_server_finalize(fd_io_server);
	err = rp_comm_tcp_server_finalize(fd_hsk_server);

	FLOG_OUT(fname);
	return NULL;
}

void rp_rpointd_manager_io(int fd, short event, void *arg)
{
	static char fname[] = "rp_rpointd_manager_io";
	static char msg[] = "fatal error";
	static Err err;
	static OpCode code;
	struct sockaddr_in addr;
	int fd_io_client;
	int flag;
	FLOG_IN(fname);
	
	switch (event) 
	{
		case (EV_READ) : {	// new connection request
			fd_io_client = rp_comm_accept(fd, &addr, sizeof(addr));
			rp_log(LOG_DEBUG, fname, "Get the I/O request from the client %s:%d");		// ip:port
			err = rp_rpointd_rq_push(&global.rq_client, fd_io_client, addr);
			err = rp_sem_post(&global.sem_worker);		
			break;
		}
		case (EV_TIMEOUT) : {
			/*	get the fd flag   */
			if ((flag = fcntl(fd, F_GETFL, 0)) < 0)
			{
				rp_log(LOG_FATAL, fname, "Cannot get the file descriptor flag. Error: %s", strerror(errno));
				close(fd);
				exit(127);
			}
			/*	turn on the non-block flag   */
			if (fcntl(fd, F_SETFL, flag|O_NONBLOCK) < 0)
			{
				rp_log(LOG_CRITICAL, fname, "Cannot set the file descriptor to Non-block mode. Error: %s", strerror(errno));
				close(fd);
				exit(127);		
			}
			/*	read from the connected client   */
			if (read(*(int*)arg, &code, sizeof(code)) == sizeof(code))
			{
				switch (code)
				{
					case (OP_END) : {		// client finished the session
						rp_log(LOG_DEBUG, fname, "The client has finished the session and disconnected.");
						err = rp_score_set_state(global.score, SS_IDLE);
						event_del((struct event *)arg);
						break;
					}
					case (OP_SHUTDOWN) : {
						int i;
						for (i=0; i<config.num_thread_prestart; i++)
						{	
							err = rp_rpointd_rq_push(&(global.rq_client), -1, addr);	// telling worker threads to exit
							err = rp_sem_post(&(global.sem_worker));
						}
						rp_log(LOG_DEBUG, fname, "Got shutdown command, notify the workers to wrap-up.");
						break;
					}
					default : {
						rp_log(LOG_WARNING, fname, "Invalid command.");
					}
				}
			}
			/*	turn off to blocking mode again   */
			if (fcntl(fd, F_SETFL, flag) < 0)
			{
				rp_log(LOG_CRITICAL, fname, "Cannot set the file descriptor to Non-block mode. Error: %s", strerror(errno));
				close(fd);
				exit(127);		
			}			
			break;
		}
	}
	FLOG_OUT(fname);
}

